package ai.people.netmon.userconfig.service;

import ai.people.core.feign.api.facebook.FacebookProbeUserClient;
import ai.people.core.feign.api.twitter.TwitterProbeUserClient;
import ai.people.core.feign.api.youtube.YoutubeProbeUserClient;
import ai.people.netmon.framework.businessenum.AnalysisSourceEnum;
import ai.people.netmon.framework.businessenum.ImportanceDegEnum;
import ai.people.netmon.framework.businessenum.gather.GatherStatusEnum;
import ai.people.netmon.framework.domain.facebook.probe.entity.FacebookAccount;
import ai.people.netmon.framework.domain.gather.response.GatherResponseBody;
import ai.people.netmon.framework.domain.twitter.probe.entity.TwitterAccount;
import ai.people.netmon.framework.domain.userconfig.entity.DataSync;
import ai.people.netmon.framework.domain.userconfig.entity.UserCollectConfigAccount;
import ai.people.netmon.framework.domain.userconfig.entity.UserCollectConfigKeywords;
import ai.people.netmon.framework.domain.userconfig.request.SyncDataRequest;
import ai.people.netmon.framework.domain.youtube.probe.entity.YoutubeAccount;
import ai.people.netmon.framework.exception.enums.CommonEnum;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * @author zhenglin
 * @date 2022/3/2
 */
@Service
@Slf4j
public class DataSyncService {

    @Autowired
    DataSync dataSync;
    @Autowired
    ExecutorService pool;
    @Autowired
    RestTemplate restTemplate;

    @Autowired
    FacebookProbeUserClient facebookProbeUserClient;
    @Autowired
    TwitterProbeUserClient twitterProbeUserClient;
    @Autowired
    YoutubeProbeUserClient youtubeProbeUserClient;


    public void sendSyncKeyWords(List<UserCollectConfigKeywords> list, Consumer<UserCollectConfigKeywords> updateConsumer, UserCollectConfigKeywords keywords, Map<String, String> pidMapping) {
        pool.execute(() -> {
            Map<String, List<UserCollectConfigKeywords>> collect = list.stream().collect(Collectors.groupingBy(UserCollectConfigKeywords::getInfoType));
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            String url = dataSync.getKeywordsUrl();
            Integer partSize = dataSync.getPartSize();
            Integer retryCount = dataSync.getRetryCount();
            AtomicBoolean flag = new AtomicBoolean(false);
            collect.forEach((k, v) -> {
                List<List<UserCollectConfigKeywords>> partition = Lists.partition(v, partSize);
                for (List<UserCollectConfigKeywords> part : partition) {
                    boolean partFlag = true;
                    part.forEach(x->x.setPartnerId(pidMapping.getOrDefault(x.getCrawlKeywords(),"[]")));
                    SyncDataRequest<UserCollectConfigKeywords> body = new SyncDataRequest<>(part, k);
                    HttpEntity<SyncDataRequest<UserCollectConfigKeywords>> request = new HttpEntity<>(body, headers);
                    log.info("请求url:{} 数据条数:{} 等待返回...", url, part.size());
                    for (int i = 0; i < retryCount; i++) {
                        GatherResponseBody result = null;
                        try {
                            result = restTemplate.postForObject(url, request, GatherResponseBody.class);
                        } catch (Exception e) {
                            log.info("重试...");
                            log.error(e.getMessage(), e);
                            continue;
                        }
                        log.info("请求url:{} 数据同步请求返回:{}",url, result);
                        //更新数据为已采集状态
                        Integer status = Optional.ofNullable(result).map(GatherResponseBody::getStatus).orElse(null);
                        if (Objects.equals(status, CommonEnum.SUCCESS.getCode())) {
                            partFlag = false;
                            break;
                        }
                    }
                    if (partFlag) {
                        flag.set(true);
                    }
                }
            });
            if (!flag.get()) {
                keywords.setCollectionStatus(GatherStatusEnum.COLLECTED.getCode());
                updateConsumer.accept(keywords);
            }
        });
    }

    /**
     * 发送到不同的源
     *
     * @param accounts 账户
     */
    public void sendToDiffSource(List<UserCollectConfigAccount> accounts) {
        Map<String, List<UserCollectConfigAccount>> collect = accounts.stream().collect(Collectors.groupingBy(UserCollectConfigAccount::getInfoType));
        pool.execute(() -> sendToFacebook(collect));
        pool.execute(() -> sendToTwitter(collect));
        pool.execute(() -> sendToYoutube(collect));
    }

    private void sendToFacebook(Map<String, List<UserCollectConfigAccount>> collect){
        List<UserCollectConfigAccount> orDefault = collect.getOrDefault(AnalysisSourceEnum.facebook.name(), Collections.emptyList());
        if (CollectionUtils.isEmpty(orDefault)) {
            return;
        }
        List<FacebookAccount> accounts = new ArrayList<>();
        orDefault.forEach(x->accounts.add(new FacebookAccount(x.getAccount(),x.getAccountLink(),String.valueOf(ImportanceDegEnum.GENERALLY_IMPORTANT.getCode()))));
        facebookProbeUserClient.saveBatch(accounts);
    }

    private void sendToTwitter(Map<String, List<UserCollectConfigAccount>> collect){
        List<UserCollectConfigAccount> orDefault = collect.getOrDefault(AnalysisSourceEnum.twitter.name(), Collections.emptyList());
        if (CollectionUtils.isEmpty(orDefault)) {
            return;
        }
        List<TwitterAccount> accounts = new ArrayList<>();
        orDefault.forEach(x->accounts.add(new TwitterAccount(x.getAccount(),x.getAccountLink(),String.valueOf(ImportanceDegEnum.GENERALLY_IMPORTANT.getCode()))));
        twitterProbeUserClient.batchSaveUser(accounts);
    }

    private void sendToYoutube(Map<String, List<UserCollectConfigAccount>> collect){
        List<UserCollectConfigAccount> orDefault = collect.getOrDefault(AnalysisSourceEnum.youtube.name(), Collections.emptyList());
        if (CollectionUtils.isEmpty(orDefault)) {
            return;
        }
        List<YoutubeAccount> accounts = new ArrayList<>();
        orDefault.forEach(x->accounts.add(new YoutubeAccount(x.getAccount(),x.getAccountLink(),String.valueOf(ImportanceDegEnum.GENERALLY_IMPORTANT.getCode()))));
        youtubeProbeUserClient.batchSaveUser(accounts);
    }

}
